package com.surfilter.massdata.spark.task;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import scala.Tuple2;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.surfilter.massdata.spark.bean.BR2002;
import com.surfilter.massdata.spark.bean.IpBean;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.ConvertUtil;
import com.surfilter.massdata.spark.util.Global;
import com.surfilter.massdata.spark.util.IpAllocateType;
import com.surfilter.massdata.spark.util.IpCheckConstants;
import com.surfilter.massdata.spark.util.iPCheckZjUtil;

/**
 * 
 * @author weimin Function:1.分别从load备案、上报库中的IP来源和分配信息 2.IP格式监测、IP段监测、IP分配属性监测
 *         3.监测之后的数据入HDFS
 * @Date:2017-05-15
 *
 */
public class IpCheckTask implements ISparkTask {

	private static final Log log = LogFactory.getLog(IpCheckTask.class);
	private static final long serialVersionUID = 1L;

	private Map<String,String> filterBbdws = Maps.newHashMap();
	private Map<String,String> needOuterBbdws = Maps.newHashMap();
	private HashSet<Long> innerProvinceBbdws = null;
	private String PROVINCE_PREFFIX = null;
	private String shengid_config = null;
	
	public List<IpBean> ipSegErrList = Lists.newArrayList();
	public List<IpBean> ipAllocateAttrErrList = Lists.newArrayList();
	public List<IpBean> beianAllocateList = Lists.newArrayList();
	public List<IpBean> oldBaseList = Lists.newArrayList();
	public List<IpBean> allIpSegList = Lists.newArrayList();
	public List<BR2002> list_bean_final = Lists.newArrayList();
	
	private static String outName;
	private String sql_2002;
	private String sql_ipBaxx;

	private static JavaSparkContext ctx;
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,
			Map<String, String> commandMap) {
		log.info("====================IpCheckGenTask begin====================");

		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try {
			//IpBean newiB = null;
			System.out.println("第一阶段--------");
			// 读取BR2002表内容
			DataFrame BR2002 = dataFrames.get(sql_2002);
			ctx = new JavaSparkContext(BR2002.sqlContext().sparkContext());
			JavaRDD<IpBean> rdd_BR2002=iPCheckZjUtil.getIpBeanRdd(BR2002);//获取基准表(RDD)
			//DataFrame result_BR2002 = BR2002.sqlContext().sql(sql_2002);
			//ctx = new JavaSparkContext(result_BR2002.sqlContext().sparkContext());
			//JavaRDD<IpBean> rdd_BR2002 = BR2002(result_BR2002, newiB);

			// 读取IP_GJCX_BAXX表内容
			DataFrame IP_GJCX_BAXX = dataFrames.get(sql_ipBaxx);
			JavaRDD<IpBean> rdd_IP_GJCX_BAXX=iPCheckZjUtil.getIpBeanRdd(IP_GJCX_BAXX);//获取备案表(RDD)
			//DataFrame result_IP_GJCX_BAXX = IP_GJCX_BAXX.sqlContext().sql(sql_ipBaxx);
			//JavaRDD<IpBean> rdd_IP_GJCX_BAXX = IP_GJCX_BAXX(result_IP_GJCX_BAXX, newiB);
			
			JavaRDD<IpBean> combineRDD=rdd_BR2002.union(rdd_IP_GJCX_BAXX);//合并2个表的数据
			System.out.println("=============combineRDD分区数为："+combineRDD.getNumPartitions());
			log.info("====================IpCheckGenTask end====================");
			
			log.info("====================IpCheckAnalyzTask begin====================");
			System.out.println("第二阶段--------");
			JavaRDD<IpBean> list_br = equleMap(combineRDD);//分析ip
			System.out.println("=============list_br分区数为："+list_br.getNumPartitions());
			log.info("====================IpCheckAnalyzTask end====================");
			
			DataFrame df = BR2002.sqlContext().createDataFrame(list_br, IpBean.class);
			outList.add(new OutQueueEntity(outName, df));
		} catch (Exception e) {
			System.err.println("data err："+iPCheckZjUtil.getLineInfo());
			e.printStackTrace();
		}
		return outList;
	}

	
	
	// 相当于MR的map阶段，将key拆分
	private JavaRDD<IpBean> equleMap(JavaRDD<IpBean> lines) {
		
		String strFilter = Global.getConfig("surfilter.ipcheck.stard.filter.bbdw");
		if(StringUtils.isNoneBlank(strFilter)){
			String[] filterBbdwsString = strFilter.split(",");//获取配置的应过滤的报备单位
			if (null != filterBbdwsString) {
				for (String id : filterBbdwsString) {
					filterBbdws.put(id,id);
				}
			}
		}
		String strOuter = Global.getConfig("surfilter.ipcheck.stard.need.outer.bbdw");
		if(StringUtils.isNoneBlank(strOuter)){
			String[] needOuterBbdwsString = strOuter.split(",");// 获取配置的应保留的外省报备单位
			if (null != needOuterBbdwsString) {
				for (String id : needOuterBbdwsString) {
					needOuterBbdws.put(id,id);
				}
			}
		}
		innerProvinceBbdws = Sets.newHashSet();
		shengid_config = ConfigUtil.get("surfilter.massdata.shengid");// 获取配置的省编码
		PROVINCE_PREFFIX = shengid_config.substring(0, 2); // 320000-->32
		innerProvinceBbdws=iPCheckZjUtil.getBr3010IdMap(shengid_config);
		
		JavaRDD<IpBean> rdd = lines.flatMapToPair(new PairFlatMapFunction<IpBean, Long, IpBean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<Tuple2<Long, IpBean>> call(IpBean t)throws Exception {
				List<Tuple2<Long, IpBean>> result = new ArrayList<Tuple2<Long, IpBean>>();
				long qsip = t.getQsip();
				long zzip = t.getZzip();
				while (qsip <= zzip) {
					result.add(new Tuple2<Long, IpBean>(qsip, t));
					qsip++;
				}
				return result;
			}
			//JavaPairRDD<Long, Iterable<IpBean>> secondRdd = firstRdd.groupByKey();
		}).groupByKey().mapToPair(new PairFunction<Tuple2<Long, Iterable<IpBean>>, String, String>() {
		//firstRdd.persist(StorageLevel.MEMORY_AND_DISK_SER());
		//JavaRDD<IpBean> rdd = firstRdd.mapToPair(new PairFunction<Tuple2<Long, Iterable<IpBean>>, String, String>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, String> call(Tuple2<Long, Iterable<IpBean>> tuple2)throws Exception {
				Long ip = tuple2._1();
				Iterable<IpBean> its = tuple2._2;
				//System.out.println("进入：IpCheckAnalyzTask>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
				Tuple2<String, String> result = IpCheckAnalyzTask(ip, its);// 分析ip
				return result;
				//JavaPairRDD<String, Iterable<String>> rdd2 = rdd1.groupByKey();
			}
		}).groupByKey().flatMap(new FlatMapFunction<Tuple2<String, Iterable<String>>, IpBean>() {
		//JavaPairRDD<String, Iterable<String>> thirdRdd = secondRdd.groupByKey();
		
		//System.out.println(">>>>>>>>>>>>>>"+thirdRdd.collect());
		//JavaRDD<IpBean> rdd = thirdRdd.flatMap(new FlatMapFunction<Tuple2<String, Iterable<String>>, IpBean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<IpBean> call(Tuple2<String, Iterable<String>> tuple2)throws Exception {
				List<IpBean> resultList = Lists.newArrayList();
				IpBean ipBean = new IpBean();
				try{
					if (StringUtils.isNotBlank(tuple2._1)) {
						String ipBeanstr = tuple2._1;
						ConvertUtil.convertIpBean(ipBeanstr.split("\t"), ipBean);
						
						//System.out.println("开始合并ip");
						List<Map<String,Object>> segListMap = iPCheckZjUtil.getIpSeg(tuple2._2);// 合并ip
						
						for (Map<String,Object> segMap : segListMap) {
							IpBean copy_ib = new IpBean();
							copy_ib = ConvertUtil.copy(copy_ib, ipBean);
							copy_ib.setQsip(Long.parseLong(String.valueOf(segMap.get("qsip"))));
							copy_ib.setZzip(Long.parseLong(String.valueOf(segMap.get("zzip"))));
							copy_ib = IpBean.copyipBean(copy_ib,(IpBean)segMap.get("ipBean"));
							
							resultList.add(copy_ib);
						}
					}
					//System.err.println("resultList最后的大小"+resultList.size());
					/*for(IpBean i : resultList){
			    	System.err.println("resultList最后的结果"+i.toString());
			    }*/
				}catch(Exception e){
					System.err.println("有数组越界情况："+iPCheckZjUtil.getLineInfo());
				}
				return resultList;
			}

		});// 得到最终结果
		
		//System.out.println("rdd结果："+rdd.collect());
		//System.out.println("第三阶段--------");
		//JavaRDD<BR2002> result_list = IpCheckCleanTask(dataSource,rdd);
		
		//return result_list;
		return rdd;
	}
	
	
	/** 第二阶段：分析ip */
	private Tuple2<String, String> IpCheckAnalyzTask(long qsip,Iterable<IpBean> iterator) {
		
		/**
		 * 分析融合ip数据
		 */
		Tuple2<String, String> result = new Tuple2<String, String>(null,null);
		IpBean br2002Ip = null;
		List<IpBean> allocateList = Lists.newArrayList();
		IpBean newestIpBean = null;
		String sumStr = null;
		boolean shouldFilter = false;
		IpBean specialReAllocateBean = null; // 再分配给了形如蓝汛这样的特殊企业
		IpBean outerBean = null; // 再分配到外省接入商

		// 遍历结果
		for (IpBean iterBean : iterator) {
			
			IpBean Link_ip = new IpBean();
			Link_ip = ConvertUtil.copy(Link_ip, iterBean);

			/** 筛选基准库IP，保留ip */
			//System.err.println("分配方式："+Link_ip.getFpfs());
			//System.err.println("分配对象："+Link_ip.getFpdx());
			//System.out.println("标志："+Link_ip.getStorageType());
			if (IpCheckConstants.OLD_BASE.equals(Link_ip.getStorageType())) {//5
				br2002Ip = Link_ip;
				
				/** 筛选备案IP */
			} else if (IpCheckConstants.BEIAN.equals(Link_ip.getStorageType())) {//1
				// 如果分配方式是"再分配"
				if (IpAllocateType.RE_ALLOCATE.getTypeValue().equals(Link_ip.getFpfs())) {//2
					if (!innerProvinceBbdws.contains(Link_ip.getFpdx()) && !needOuterBbdws.containsKey(Link_ip.getFpdx().toString())) {
						// 如果再分配中出现了外省接入商，同时又不是蓝汛这样的特殊企业 就丢掉该IP
						shouldFilter = true;
						outerBean = Link_ip;
					} else if (!innerProvinceBbdws.contains(Link_ip.getFpdx()) && needOuterBbdws.containsKey(Link_ip.getFpdx().toString())) {
						// 如果再分配中出现了外省接入商，同时是蓝汛这样的特殊企业 ,保留ip
						specialReAllocateBean = Link_ip;
						//System.err.println("特殊bean:"+specialReAllocateBean.toString());
						//break;
					}else{
						outerBean = Link_ip;
						break;
					}
				} else {
					allocateList.add(Link_ip);
				}
			}
		}
		
		/* 在保留的ip中，将对应的IP信息中，分配方式置为空，分配对象置为空，报备单位ID设置为分配对象的ID,若省非本身，则设置为本省 */
		if (specialReAllocateBean != null) {
			specialReAllocateBean.setFpfs(null);
			specialReAllocateBean.setBbdw_id(specialReAllocateBean.getFpdx());
			specialReAllocateBean.setFpdx(null);
			// 如果来自备案库的该条信息的省市县不是本省 那么手动置为本省
			if (!specialReAllocateBean.getShengid().toString().startsWith(PROVINCE_PREFFIX)) {
				specialReAllocateBean.setShengid(Long.valueOf(shengid_config));
				// specialReAllocateBean.setShiid(330100l);
			}
			try {
				long sbsj = specialReAllocateBean.getSbsj();
				//specialReAllocateBean.setSbsj(0L);
				sumStr = sbsj + "\001" + qsip + "\001" + specialReAllocateBean.toString();
				String newIb = specialReAllocateBean.simpleBean();
				return new Tuple2<String, String>(newIb, sumStr);
				// mergeIpSeg(specialReAllocateBean, ip,);
			} catch (Exception e) {
				e.printStackTrace();
				System.err.println("返回specialReAllocateBean数据错误："+iPCheckZjUtil.getLineInfo());
			}
		}

		// TODO 测试
		/*for (IpBean aa : allocateList) {
			System.out.println(allocateList.size()+">>>");
			System.err.println("allocateList结果>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>：" + aa.toString());
		}*/
		
		/** 备案库IP与基准库IP融合比对 */
		newestIpBean = this.getNewestBeianIp(allocateList);
		
		
		//for(IpBean newestIpBean : allocateList){
			/* 若IP同时存在于备案库和基准库 */
			if (newestIpBean != null && br2002Ip != null) {
				newestIpBean.setOperatorsName(br2002Ip.getOperatorsName());
				newestIpBean.setOperatorsId(br2002Ip.getOperatorsId());
				newestIpBean.setOperatorsBh(br2002Ip.getOperatorsBh());
				newestIpBean.setHouseCode(br2002Ip.getHouseCode());
				newestIpBean.setHouseName(br2002Ip.getHouseName());
				//newestIpBean.setJtiplx(br2002Ip.getJtiplx());
				newestIpBean.setJtiplx(newestIpBean.getIplx());
				if (newestIpBean.getShengid().equals(-1l) || newestIpBean.getShiid().equals(-1l) || newestIpBean.getShengid().equals(0l)
						|| newestIpBean.getShiid().equals(0l)
						|| !newestIpBean.getShengid().toString().startsWith(PROVINCE_PREFFIX)
						|| !newestIpBean.getShiid().toString().startsWith(PROVINCE_PREFFIX)
						|| newestIpBean.getShengid() == null || newestIpBean.getShiid() == null) {
					newestIpBean.setShengid(br2002Ip.getShengid());
					newestIpBean.setShiid(br2002Ip.getShiid());
					newestIpBean.setXianid(br2002Ip.getXianid());
				}
				
				try {
					long sbsj = newestIpBean.getSbsj();
					//newestIpBean.setSbsj(0L);
					sumStr = sbsj + "\001" + qsip + "\001" + newestIpBean.toString();
					String newIb = newestIpBean.simpleBean();
					return new Tuple2<String, String>(newIb.toString(),sumStr);
					// mergeIpSeg(newestIpBean, ip,);
				} catch (Exception e) {
					e.printStackTrace();
					System.err.println("返回newestIpBean1数据错误:"+iPCheckZjUtil.getLineInfo());
				}
				
				/* 若IP只存在于备案库 */
			} else if (newestIpBean != null && br2002Ip == null) {
				newestIpBean.setJtiplx(newestIpBean.getIplx());
				if (!newestIpBean.getShengid().toString().startsWith(PROVINCE_PREFFIX) || !newestIpBean.getShiid().toString().startsWith(PROVINCE_PREFFIX)) {
					// 备案库里的IP信息为外省 那么就全置为空
					newestIpBean.setShengid(0l);
					newestIpBean.setShiid(0l);
					newestIpBean.setXianid(0l);
				}
				try {
					long sbsj = newestIpBean.getSbsj();
					//newestIpBean.setSbsj(0L);
					sumStr = sbsj + "\001" + qsip + "\001" + newestIpBean.toString() ;
					String newIb = newestIpBean.simpleBean();
					return new Tuple2<String, String>(newIb,sumStr);
					// mergeIpSeg(newestIpBean, ip,);
				} catch (Exception e) {
					e.printStackTrace();
					System.out.println("返回newestIpBean2数据错误");
				}
				/* 若IP只存在于基准库 */
			} else if (newestIpBean == null && br2002Ip != null) {
				try {
					// mergeIpSeg(br2002Ip, ip, context);
					/*long sbsj = br2002Ip.getSbsj();
					br2002Ip.setSbsj(null);
					sumStr = sbsj + "," + qsip;
					return new Tuple2<String, String>(br2002Ip.toString(),sumStr);*/
					return result;
				} catch (Exception e) {
					e.printStackTrace();
					System.err.println("返回数据错误:"+iPCheckZjUtil.getLineInfo());
				}
			}
		//}
	
		return result;
	}


	// 在key相同的情况下，取时间最新的value
	public IpBean getNewestBeianIp(List<IpBean> allocateList) {
		IpBean ipBean = null;
		if (allocateList.size() == 0)
			return ipBean;
		if (allocateList.size() == 1)
			return allocateList.get(0);

		for (IpBean iterBean : allocateList) {
			if (filterBbdws.containsKey(iterBean.getBbdw_id().toString())) {
				continue;
			}
			if (ipBean == null) {
				ipBean = iterBean;
			} else {
				if (iterBean.getSbsj().compareTo(ipBean.getSbsj()) > 0) {
					ipBean = iterBean;
				}
			}
		}
		return ipBean;
	}

	
}
